package cn.ac.iie.di.ban.ycl.cleaner.sender;

import cn.ac.iie.di.ban.ycl.cleaner.CleanerMain;
import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import java.util.List;
import org.apache.commons.configuration2.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn>
 * @date 2016-11-8 11:14:11
 * @version V1.0
 * @desc
 */
public class RubbishMessageSenderThread implements Runnable {

    private static final Logger LOG = LoggerFactory.getLogger(RubbishMessageSenderThread.class);
    MessageSender sender;

    public RubbishMessageSenderThread(Configuration conf) {
        String nameServer = conf.getString(Constants.RECEIVER_MQ_NAMESERVER);
        String topic = conf.getString(Constants.RUBBISH_SEND_TOPIC);
        sender = new MessageSender(nameServer, topic);
    }

    @Override
    public void run() {
        LOG.info("rubbish send thread start!");
        while (true) {
            try {
                REMessage message = (REMessage) CleanerMain.rubbishQueue.take();
                if (message == null) {
                    continue;
                }
                List<Object[]> datas = message.data;
                if (datas == null || datas.isEmpty()) {
                    continue;
                }
                sender.sendMessage(message);
                LOG.info("send rubbish messages num: {}", datas.size());
            } catch (Exception ex) {
                LOG.error("send rubbish messages error!", ex);
            }
        }
    }

}
